package ali;

import com.google.common.util.concurrent.ThreadFactoryBuilder;

import java.util.LinkedList;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.*;

/**
 * @author wei.yang on 2018/8/27
 */
public class Main {

    private static ThreadFactory factory = new ThreadFactoryBuilder().setNameFormat("THREAD-%d").build();

    private static ExecutorService cosumerService = new ThreadPoolExecutor(5, 10, 0, TimeUnit.SECONDS, new LinkedBlockingQueue<>(), factory);

    private static ExecutorService producerService = new ThreadPoolExecutor(2, 5, 0, TimeUnit.SECONDS, new LinkedBlockingQueue<>(), factory);

    public static void main(String[] args) {
        Queue<String> queue = new LinkedList<>();
        final Object object = new Object();
        while (true) {
            cosumerService.execute(new Consumer(queue, object));
            producerService.execute(new Producer(queue, object));
        }
    }


    public static class Consumer implements Runnable {

        private Queue<String> queue;

        private final Object lock;

        Consumer(Queue<String> queue, final Object lock) {
            this.queue = queue;
            this.lock = lock;
        }

        @Override
        public void run() {
            String message = queue.poll();
            synchronized (lock) {
                if (message == null || "".equals(message)) {
                    try {
                        lock.wait(100);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    return;
                }
            }
            boolean success = MessageManager.dealOne(message);
            if (!success) {
                queue.add(message);
            }
        }
    }

    public static class Producer implements Runnable {

        private Queue<String> queue;

        private final Object lock;

        Producer(Queue<String> queue, final Object lock) {
            this.queue = queue;
            this.lock = lock;
        }

        @Override
        public void run() {
            synchronized (lock) {
                if (queue.size() > 1000) {
                    lock.notify();
                }
            }
            List<String> messages = MessageManager.readBatch();
            if (messages == null || messages.size() == 0) {
                return;
            }
            queue.addAll(messages);
        }
    }
}

